﻿using kratos;
using RingBuffer;
using System;
using System.Buffers;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;

namespace kratos {

/// <summary>
/// 网络事件ID
/// </summary>
internal enum NetEventID {
  none = 0,
  connect_request = 1,
  connect_ack,
  connect_fail,
  reconnect_ack,
  close_request,
  close_ack,
  close_notify,
  send_request,
  send_ack,
  send_fail,
  recv_notify,
  recv_fail,
  recv_timeout,
  peer_close,
  shutdown,
}

/// <summary>
/// 网络事件接口
/// </summary>
internal interface NetEventBase {
  /// <summary>
  /// 获取事件ID
  /// </summary>
  /// <returns>事件ID</returns>
  NetEventID EventID { get; }
}

internal class NetEvent : NetEventBase {
  NetEventID id_ = NetEventID.none;
  ulong session_id_ = 0;
  public NetEventID EventID {
    set { id_ = value; }
    get { return id_; }
  }
  public ulong SessionID {
    get { return session_id_; }
    set { session_id_ = value; }
  }
}

internal class NetEventData : NetEvent, IRecycle {
  byte[] buffer_ = null;
  int length_ = 0;
  int max_length_ = 0;
  public byte[] Data {
    get { return buffer_; }
  }
  public int DataLength {
    get { return length_; }
  }
  public void write(byte[] data, int size) {
    if (max_length_ < size) {
      max_length_ = size;
      if (buffer_ != null) {
        GlobalBytePool.Pool.Return(buffer_);
      }
      buffer_ = GlobalBytePool.Pool.Rent(size);
    }
    length_ = size;
    data.CopyTo(buffer_, 0);
  }
  public int read(GrowingRingBuffer<byte> rb) {
    for (int i = 0; i < length_; i++) {
      rb.Put(buffer_[i]);
    }
    return length_;
  }

  public void OnReset() {
    if (buffer_ != null) {
      GlobalBytePool.Pool.Return(buffer_);
    }
    buffer_ = null;
    length_ = 0;
    max_length_ = 0;
  }
}

internal class NetEventConnect : NetEvent {
  string host_ = null;
  int port_ = 0;
  public string Host {
    set { host_ = value; }
    get { return host_; }
  }
  public int Port {
    set { port_ = value; }
    get { return port_; }
  }
}

internal enum ClientState {
  none = 0,
  wait_connect,
  wait_reconnect,
  connecting,
  reconnecting,
  connected,
  disconnected,
}

internal class Worker {
  IClientHandler handler_ = null;
  Client client_ = null;
  Socket socket_ = null;
  ulong session_id_ = 0;
  Thread worker_ = null;
  bool running_ = false;
  string host_ = null;
  int port_ = 0;
  ArrayList select_fd_connect = new ArrayList();
  ArrayList select_fd_read = new ArrayList();
  ArrayList select_fd_write = new ArrayList();
  ConcurrentQueue<NetEventBase> main_worker_queue_ =
      new ConcurrentQueue<NetEventBase>();
  ConcurrentQueue<NetEventBase> worker_main_queue_ =
      new ConcurrentQueue<NetEventBase>();
  GrowingRingBuffer<byte> recv_buffer_ =
      new GrowingRingBuffer<byte>(1024 * 1024);
  GrowingRingBuffer<byte> send_buffer_ =
      new GrowingRingBuffer<byte>(1024 * 1024);
  ConcurrentQueue<Object> main_worker_object_queue_ =
      new ConcurrentQueue<Object>();
  ConcurrentQueue<Object> worker_main_object_queue_ =
      new ConcurrentQueue<Object>();
  ClientState state_ = ClientState.none;
  byte[] temp_recv_buffer_ = null;
  byte[] temp_write_buffer_ = null;
  long timeout_ = 5000;
  long recv_timeout_ = 5000;
  long recv_deadline_ = 0;
  long connect_deadline_ = 0;
  Encryptor encryptor_ = null;
  Serializer serializer_ = null;
  static ulong session_id_start_ = 1;
  static long get_now_ms() {
    return (DateTime.Now.ToUniversalTime().Ticks - 621355968000000000) / 10000;
  }
  public Worker(Client client, long timeout, long recv_timeout,
                IClientHandler handler) {
    handler_ = handler;
    client_ = client;
    timeout_ = timeout;
    recv_timeout_ = recv_timeout;
  }
  public Serializer ObjectSerializer {
    get { return serializer_; }
    set { serializer_ = value; }
  }
  public Encryptor Cryptor {
    get { return encryptor_; }
    set { encryptor_ = value; }
  }
  public bool Connected {
    get { return (state_ == ClientState.connected); }
  }
  public bool connect(string host, int port) {
    if (worker_ == null) {
      running_ = true;
      worker_ = new Thread(Worker.worker_proc);
      worker_.Start(this);
    }
    connect_deadline_ = get_now_ms() + timeout_;
    NetEventConnect event_connect = new NetEventConnect();
    event_connect.Host = host;
    event_connect.Port = port;
    event_connect.EventID = NetEventID.connect_request;
    main_worker_queue_.Enqueue(event_connect);
    return true;
  }
  public bool reconnect(string host, int port) {
    if (state_ == ClientState.disconnected) {
      return connect(host, port);
    } else {
      close();
      return connect(host, port);
    }
  }
  private void on_event() {
    NetEventBase event_base = null;
    NetEventData data = null;
    while (worker_main_queue_.TryDequeue(out event_base)) {
      // 处理网路线程发送来的事件
      switch (event_base.EventID) {
      case NetEventID.recv_notify:
        data = (NetEventData) event_base;
        data.read(recv_buffer_);
        ObjectPool<NetEventData>.Recycle(data);
        if (handler_) {
          handler_.OnReceive(client_, recv_buffer_);
        }
        break;
      case NetEventID.close_ack:
      case NetEventID.close_notify:
      case NetEventID.peer_close:
        if (handler_) {
          handler_.OnClose(client_);
        }
        break;
      case NetEventID.send_fail:
        if (handler_) {
          handler_.OnSendFail(client_);
        }
        break;
      case NetEventID.send_ack:
        break;
      case NetEventID.connect_fail:
        if (handler_) {
          handler_.OnConnectFail(client_);
        }
        break;
      case NetEventID.connect_ack:
        session_id_ = ((NetEvent) event_base).SessionID;
        if (handler_) {
          handler_.OnConnect(client_);
        }
        break;
      case NetEventID.reconnect_ack:
        if (handler_) {
          handler_.OnReconnect(client_);
        }
        break;
      case NetEventID.recv_timeout:
        if (handler_) {
          handler_.OnReceiveTimeout(client_);
        }
        break;
      default:
        break;
      }
    }
  }
  private void on_recv_object() {
    Object obj = null;
    while (worker_main_object_queue_.TryDequeue(out obj)) {
      if (handler_) {
        handler_.OnReceiveObject(client_, obj);
      }
    }
  }
  public void update() {
    long now = get_now_ms();
    if (!Connected) {
      if (now > connect_deadline_) {
        if (handler_) {
          handler_.OnConnectFail(client_);
        }
        close();
      }
    }
    on_event();
    on_recv_object();
  }
  public void close() {
    NetEvent net_event = new NetEvent();
    net_event.EventID = NetEventID.close_request;
    net_event.SessionID = session_id_;
    main_worker_queue_.Enqueue(net_event);
  }

  public int send(byte[] data, int size) {
    NetEventData net_event = ObjectPool<NetEventData>.Get();
    net_event.EventID = NetEventID.send_request;
    net_event.SessionID = session_id_;
    net_event.write(data, size);
    main_worker_queue_.Enqueue(net_event);
    return size;
  }

  public bool send(Object obj) {
    main_worker_object_queue_.Enqueue(obj);
    return true;
  }

  private bool is_running() { return running_; }

  private void stop_worker() {
    running_ = false;
    NetEvent net_event = new NetEvent();
    net_event.EventID = NetEventID.shutdown;
    main_worker_queue_.Enqueue(net_event);
  }

  public void Dispose() {
    // 关闭线程
    stop_worker();
  }
  private void on_worker_clear_buffer() {
    recv_buffer_.Clear();
    send_buffer_.Clear();
    Object temp_obj;
    while (main_worker_object_queue_.Count > 0) {

      main_worker_object_queue_.TryDequeue(out temp_obj);
    }
    while (worker_main_object_queue_.Count > 0) {

      worker_main_object_queue_.TryDequeue(out temp_obj);
    }
  }
  private void on_worker_close() {
    if (socket_ != null) {
      socket_.Close();
      socket_ = null;
    }
    on_worker_clear_buffer();
    NetEvent close_ack = new NetEvent();
    close_ack.EventID = NetEventID.close_ack;
    close_ack.SessionID = session_id_;
    worker_main_queue_.Enqueue(close_ack);
    state_ = ClientState.disconnected;
  }
  private void on_worker_close_notify() {
    on_worker_clear_buffer();
    NetEvent close_notify = new NetEvent();
    close_notify.EventID = NetEventID.close_notify;
    close_notify.SessionID = session_id_;
    worker_main_queue_.Enqueue(close_notify);
  }
  public bool on_worker_event() {
    NetEventBase event_base = null;
    while (main_worker_queue_.TryDequeue(out event_base)) {
      switch (event_base.EventID) {
      case NetEventID.close_request:
        on_worker_close();
        break;
      case NetEventID.send_request:
        try {
          on_worker_send((NetEventData) event_base);
        } finally {
          ObjectPool<NetEventData>.Recycle(event_base);
        }
        break;
      case NetEventID.connect_request:
        on_worker_connect((NetEventConnect) event_base);
        break;
      case NetEventID.shutdown:
        return false;
      default:
        break;
      }
    }
    return true;
  }
  private void on_worker_send(NetEventData event_data) {
    byte[] raw_data = null;
    int raw_size = 0;
    if (encryptor_ != null) {
      raw_data = encryptor_.encrypt(event_data.Data, event_data.DataLength);
      raw_size = raw_data.Length;
    } else {
      raw_data = event_data.Data;
      raw_size = event_data.DataLength;
    }
    for (int i = 0; i < raw_size; i++) {
      send_buffer_.Put(raw_data[i]);
    }
  }
  private void on_worker_connect_success() {
    NetEventID event_id = NetEventID.none;
    if (state_ == ClientState.connecting) {
      event_id = NetEventID.connect_ack;
    } else {
      event_id = NetEventID.reconnect_ack;
    }
    recv_deadline_ = get_now_ms() + recv_timeout_;
    state_ = ClientState.connected;
    NetEvent net_event = new NetEvent();
    net_event.EventID = event_id;
    net_event.SessionID = session_id_start_++;
    worker_main_queue_.Enqueue(net_event);
  }
  private void on_worker_send_success() {
    NetEvent net_event = new NetEvent();
    net_event.EventID = NetEventID.send_ack;
    net_event.SessionID = session_id_;
    worker_main_queue_.Enqueue(net_event);
  }
  private void on_worker_connect(NetEventConnect event_connect) {
    if (state_ != ClientState.none) {
      if (state_ == ClientState.connected) {
        // 已经连接完成的需要发送一个关闭通知
        NetEvent net_event = new NetEvent();
        net_event.EventID = NetEventID.close_notify;
        worker_main_queue_.Enqueue(net_event);
      }
      state_ = ClientState.wait_reconnect;
      if (socket_ != null) {
        socket_.Close();
        socket_.Dispose();
        socket_ = null;
      }
    }
    host_ = event_connect.Host;
    port_ = event_connect.Port;
    socket_ = new Socket(SocketType.Stream, ProtocolType.Tcp);
    socket_.NoDelay = true;   // 关闭Nagle
    socket_.Blocking = false; // 非阻塞
    socket_.ReceiveBufferSize = 1024 * 1024;
    socket_.SendBufferSize = 1024 * 1024;
    IPEndPoint end_point = null;
    try {
      IPAddress Ip_addr = IPAddress.Parse(host_);
      end_point = new IPEndPoint(Ip_addr, port_);
    } catch (Exception) {
      worker_close_socket(NetEventID.connect_fail);
      return;
    }
    if (state_ == ClientState.none) {
      state_ = ClientState.wait_connect;
    }
    // 连接服务器
    try {
      socket_.Connect(end_point);
    } catch (SocketException) {
      // 异步等待连接成功
      if (state_ == ClientState.wait_connect) {
        state_ = ClientState.connecting;
      } else if (state_ == ClientState.wait_reconnect) {
        state_ = ClientState.reconnecting;
      } else {
        worker_close_socket(NetEventID.connect_fail);
        return;
      }
      return;
    } catch (Exception) {
      worker_close_socket(NetEventID.connect_fail);
      return;
    }
    // 直接连接成功
    on_worker_connect_success();
  }
  private void worker_close_socket(NetEventID event_id) {
    recv_buffer_.Clear();
    send_buffer_.Clear();
    NetEvent net_event = new NetEvent();
    net_event.EventID = event_id;
    net_event.SessionID = session_id_;
    worker_main_queue_.Enqueue(net_event);
    if (state_ == ClientState.connected) {
      on_worker_close_notify();
    }
    if (socket_ != null) {
      socket_.Close();
      socket_ = null;
    }
    state_ = ClientState.disconnected;
  }
  private void worker_recv(byte[] data, int size) {
    byte[] raw_data = null;
    int raw_size = 0;
    if (encryptor_ != null) {
      raw_data = encryptor_.decrypt(data, size);
      if (raw_data == null) {
        // 没有接收完全，等待下次接收
        return;
      }
      raw_size = raw_data.Length;
    } else {
      raw_data = data;
      raw_size = size;
    }
    if (serializer_ != null) {
      Object obj = serializer_.deserialize(raw_data, raw_size);
      while (obj != null) {
        // 发送给主线程
        worker_main_object_queue_.Enqueue(obj);
        obj = serializer_.deserialize(null, 0);
      }
      return;
    }
    NetEventData event_data = ObjectPool<NetEventData>.Get();
    event_data.SessionID = session_id_;
    event_data.EventID = NetEventID.recv_notify;
    event_data.write(raw_data, raw_size);
    worker_main_queue_.Enqueue(event_data);
  }
  private void on_worker_select() {
    if ((state_ == ClientState.connecting) ||
        (state_ == ClientState.reconnecting)) {
      select_fd_connect.Clear();
      select_fd_connect.Add(socket_);
      Socket.Select(null, select_fd_connect, null, 1000);
      if (select_fd_connect.Count > 0) {
        // 连接成功
        on_worker_connect_success();
      }
    } else if (state_ == ClientState.connected) {
      if (recv_deadline_ > get_now_ms()) {
        recv_deadline_ += recv_timeout_;
        NetEvent net_event = new NetEvent();
        net_event.EventID = NetEventID.recv_timeout;
        worker_main_queue_.Enqueue(net_event);
      }
      select_fd_write.Clear();
      select_fd_read.Clear();
      if (send_buffer_.Count > 0) {
        select_fd_write.Add(socket_);
      }
      select_fd_read.Add(socket_);
      if (select_fd_write.Count > 0) {
        Socket.Select(select_fd_read, select_fd_write, null, 1000);
      } else {
        Socket.Select(select_fd_read, null, null, 1000);
      }
      for (int i = 0; i < select_fd_read.Count; i++) {
        int size = socket_.Available;
        if (size > 0) {
          if (temp_recv_buffer_ == null) {
            temp_recv_buffer_ = new byte[size];
          } else {
            if (size > temp_recv_buffer_.Length) {
              temp_recv_buffer_ = new byte[size];
            }
          }
          try {
            int recv_bytes =
                socket_.Receive(temp_recv_buffer_, size, SocketFlags.None);
            if (0 == recv_bytes) {
              worker_close_socket(NetEventID.peer_close);
            } else {
              worker_recv(temp_recv_buffer_, recv_bytes);
            }
            recv_deadline_ += recv_timeout_;
          } catch (Exception) {
            worker_close_socket(NetEventID.recv_fail);
          }
        }
      }
      for (int i = 0; i < select_fd_write.Count; i++) {
        int size = send_buffer_.Count;
        if (temp_write_buffer_ == null) {
          temp_write_buffer_ = new byte[size];
        } else {
          if (size > temp_write_buffer_.Length) {
            temp_write_buffer_ = new byte[size];
          }
        }
        send_buffer_.CopyTo(temp_write_buffer_, 0);
        try {
          int send_bytes =
              socket_.Send(temp_write_buffer_, size, SocketFlags.None);
          if (size == send_bytes) {
            // 全部发送完毕
            send_buffer_.Clear();
          } else {
            // 发送了一部分
            while (send_bytes > 0) {
              send_buffer_.Get();
              send_bytes -= 1;
            }
          }
          on_worker_send_success();
        } catch (Exception) {
          worker_close_socket(NetEventID.send_fail);
        }
      }
    } else {
      Thread.Sleep(1);
    }
  }

  private void on_object_send() {
    Object obj = null;
    while (main_worker_object_queue_.TryDequeue(out obj)) {
      if (serializer_ == null) {
        // 错误
        worker_close_socket(NetEventID.send_fail);
        return;
      }
      byte[] data = serializer_.serialize(obj);
      if (data == null) {
        // 错误
        worker_close_socket(NetEventID.send_fail);
        return;
      }
      byte[] raw_data = null;
      int size = 0;
      if (encryptor_ == null) {
        raw_data = data;
        size = data.Length;
      } else {
        raw_data = encryptor_.encrypt(data, data.Length);
        size = raw_data.Length;
        if (raw_data == null) {
          // 错误
          worker_close_socket(NetEventID.send_fail);
          return;
        }
      }
      NetEventData event_data = ObjectPool<NetEventData>.Get();
      event_data.EventID = NetEventID.send_request;
      event_data.write(raw_data, size);
      try {
        on_worker_send(event_data);
      } finally {
        ObjectPool<NetEventData>.Recycle(event_data);
      }
    }
  }
  private void terminate() {
    if (socket_ != null) {
      if (socket_.Connected) {
        socket_.Close();
      }
      socket_.Dispose();
      socket_ = null;
    }
  }
  public static void worker_proc(Object obj) {
    Worker worker = (Worker) obj;
    while (worker.is_running()) {
      try {
        if (!worker.on_worker_event()) {
          break;
        }
        worker.on_worker_select();
        worker.on_object_send();
      } finally {
      }
    }
    worker.terminate();
  }
}

/// <summary>
/// TCP客户端
/// 连接到远程服务器并与之通信，支持多种功能
/// 1. 连接超时
/// 2. 读超时
/// 3. 网络事件回调
/// 4. 加解密HOOK
/// 5. 对象序列化/反序列HOOK
/// </summary>
internal class Client : IClient {
  /// <summary>
  /// 网络工作线程
  /// </summary>
  Worker worker_ = null;
  /// <summary>
  /// 远程主机地址
  /// </summary>
  string host_ = null;
  /// <summary>
  /// 远程主机端口
  /// </summary>
  int port_ = 0;
  /// <summary>
  /// 是否由用户主动关闭
  /// </summary>
  bool manual_closed_ = false;
  public bool Connect(string host, int port, long timeout, long recv_timeout,
                      IClientHandler handler) {
    manual_closed_ = false;
    if (worker_ != null) {
      return false;
    }
    host_ = host;
    port_ = port;
    worker_ = new Worker(this, timeout, recv_timeout, handler);
    return worker_.connect(host, port);
  }
  public bool Connect(string host, int port, IClientHandler handler) {
    return Connect(host, port, 5000, 5000, handler);
  }
  public Serializer ObjectSerializer {
    get { return worker_.ObjectSerializer; }
    set { worker_.ObjectSerializer = value; }
  }
  public Encryptor Cryptor {
    get { return worker_.Cryptor; }
    set { worker_.Cryptor = value; }
  }

  public bool Reconnect() {
    manual_closed_ = false;
    if (worker_ != null) {
      return worker_.reconnect(host_, port_);
    } else {
      return false;
    }
  }

  public bool Connected {
    get {
      if (manual_closed_ || (worker_ == null)) {
        return false;
      } else {
        return worker_.Connected;
      }
    }
  }

  public void Close() {
    if (!Connected) {
      return;
    }
    manual_closed_ = true;
    worker_.close();
  }

  public void Update() {
    if (worker_ != null) {
      worker_.update();
    }
  }

  public int Send(byte[] data, int size) {
    if (manual_closed_) {
      return 0;
    }
    if (data == null || size == 0) {
      return 0;
    }
    return worker_.send(data, size);
  }

  public bool Send(Object obj) {
    if (manual_closed_ || (obj == null)) {
      return false;
    }
    if (worker_.ObjectSerializer == null) {
      return false;
    }
    return worker_.send(obj);
  }
  public void Dispose() {
    if (worker_ != null) {
      worker_.Dispose();
    }
    worker_ = null;
  }
}
}
